package com.cx.service;

import com.cx.support.RedisBatch;
import com.cx.support.RedisBatchExecute;
import com.cx.support.RedisSingle;
import com.cx.utils.Const;
import org.apache.commons.lang3.StringUtils;
import org.springframework.dao.DataAccessException;
import org.springframework.data.redis.core.RedisOperations;
import org.springframework.data.redis.core.SessionCallback;
import org.springframework.util.CollectionUtils;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.Pipeline;
import redis.clients.jedis.Transaction;
import redis.clients.util.Pool;

import java.io.IOException;
import java.util.List;

/**
 * 采用Pipeline管道模式提高批量插入效率
 * Created on 2019-04-18 15:42.
 * @author Alan Shu
 */
public class RedisTransaction extends RedisTemplateAdapter {

    /**
     * 批量执行（无事务，无乐观锁，重试5次）
     * @param redisBatch
     * @return
     */
    public static List<Object> batchExecute(RedisBatch redisBatch) {
        return batchExecute(redisBatch, false, null);
    }

    /**
     * 批量执行（无事务，有乐观锁，重试5次）
     * @param redisBatch
     * @param lockCacheKey
     * @return
     */
    public static List<Object> batchExecute(RedisBatch redisBatch, String lockCacheKey) {
        return batchExecute(redisBatch, false, lockCacheKey);
    }

    /**
     * 批量执行（是否支持事务、乐观锁，重试5次）
     *
     * @param redisBatch
     */
    public static List<Object> batchExecute(RedisBatch redisBatch, boolean enableTx, String lockCacheKey) {
        Transaction tx = null;
        List<Object> result = null;
        for(int index = 1; index <= Const.RETRY_TIMES ; index++) {
            Pool<Jedis> pool =  RedisTemplateAdapter.getPool();
            Jedis jedis = pool.getResource();
            try {
                Pipeline pipeline = jedis.pipelined();
                if (StringUtils.isNotBlank(lockCacheKey)) {
                    pipeline.watch(lockCacheKey);
                }
                if (enableTx) {
                    tx = jedis.multi();
                }

                redisBatch.call(pipeline);
                result = enableTx ? tx.exec() : pipeline.syncAndReturnAll();
            } catch (Exception e) {
                if (index < Const.RETRY_TIMES) {
                    continue;
                }
                throw e;
            } finally {
                if(jedis.getClient().isInMulti()){
                    try {
                        tx.close();
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                }
                if (jedis.getClient().isInWatch()) {
                    jedis.unwatch();
                }
                returnResource(pool, jedis);
//                closeJedis(jedis);
            }
            break;
        }
        return result;
    }

    /**
     * 单个执行查询（有返回值，无乐观锁，重试5次）
     * @param redisSingle
     * @return
     */
    public static Object singleExecute(RedisSingle redisSingle) {
        return singleExecute(redisSingle, true, null);
    }

    /**
     * 单个执行（无返回值，有乐观锁，重试5次）
     * @param redisSingle
     * @param lockCacheKey
     * @return
     */
    public static void singleExecute(RedisSingle redisSingle, String lockCacheKey) {
        singleExecute(redisSingle, false, lockCacheKey);
    }

    /**
     * 单个执行（是否支持乐观锁、返回结果，重试5次）
     *
     * @param redisSingle
     * @return
     */
    public static Object singleExecute(RedisSingle redisSingle, boolean hasRet, String lockCacheKey) {
        Object result = null;
        for(int index = 1; index <= Const.RETRY_TIMES ; index++) {
            Pool<Jedis> pool =  RedisTemplateAdapter.getPool();
            Jedis jedis = pool.getResource();
            try {
                Pipeline pipeline = jedis.pipelined();
                if (StringUtils.isNotBlank(lockCacheKey)) {
                    pipeline.watch(lockCacheKey);
                }

                redisSingle.call(pipeline);
                if(hasRet) {
                    List<Object> list = pipeline.syncAndReturnAll();
                    if (!CollectionUtils.isEmpty(list)) {
                        result = list.get(0);
                    }
                }else{
                    pipeline.sync();
                }
            } catch (Exception e) {
                if (index < Const.RETRY_TIMES) {
                    continue;
                }
                throw e;
            } finally {
                if (jedis.getClient().isInWatch()) {
                    jedis.unwatch();
                }
                returnResource(pool, jedis);
//                closeJedis(jedis);
            }
            break;
        }
        return result;
    }

    /**
     * 事务批量执行redis任务（无乐观锁，重试5次）
     * @param redisBatchExecute
     * @return
     */
    public static <T> List<T> execute(RedisBatchExecute redisBatchExecute) {
        return execute(redisBatchExecute, null);
    }

    /**
     * 事务乐观锁批量执行redis任务（重试5次）
     * @param redisBatchExecute
     * @param lockCacheKey - 乐观锁
     * @return
     */
    @SuppressWarnings({"rawtypes", "unchecked"})
    public static <T> List<T> execute(RedisBatchExecute redisBatchExecute, String lockCacheKey) {
        return CacheObjectTemplate.getRedisTemplate().execute(new SessionCallback<List<T>>() {
            @Override
            public List<Object> execute(RedisOperations operations) throws DataAccessException {
                Transaction tx = null;
                List<Object> result = null;
                for(int index = 1; index <= Const.RETRY_TIMES ; index++) {
                    Pool<Jedis> pool =  RedisTemplateAdapter.getPool();
                    Jedis jedis = pool.getResource();
                    try {
                        if (StringUtils.isNotBlank(lockCacheKey)) {
                            operations.watch(lockCacheKey);
                        }

                        tx = jedis.multi();
                        redisBatchExecute.call(operations);
                        result = tx.exec();
                    } catch (Exception e) {
                        if (index < Const.RETRY_TIMES) {
                            continue;
                        }
                        throw e;
                    } finally {
                        if(jedis.getClient().isInMulti()){
                            try {
                                tx.close();
                            } catch (IOException e) {
                                e.printStackTrace();
                            }
                        }
                        if (jedis.getClient().isInWatch()) {
                            operations.unwatch();
                        }
                        returnResource(pool, jedis);
//                        closeJedis(jedis);
                    }
                    break;
                }
                return result;
            }
        });
    }

    /**
     * 管道执行redis任务（无事务，有乐观锁，重试5次）
     * @param redisBatchExecute
     * @param lockCacheKey    - 乐观锁
     * @return 返回值被覆盖，不能有非空返回值
     */
    public static <T> List<Object> executePipelined(RedisBatchExecute redisBatchExecute, String lockCacheKey) {
        return executePipelined(redisBatchExecute, false, lockCacheKey);
    }


    /**
     * 管道批量执行redis任务（是否支持事务、乐观锁，重试5次）
     * @param redisBatchExecute
     * @param lockCacheKey - 乐观锁
     * @param enableTx 是否开启事务
     * @return 返回值被覆盖，不能有非空返回值
     */
    @SuppressWarnings({"rawtypes", "unchecked"})
    public static <T> List<Object> executePipelined(RedisBatchExecute redisBatchExecute, boolean enableTx, String lockCacheKey) {
        return CacheObjectTemplate.getRedisTemplate().executePipelined(new SessionCallback<List<T>>() {
            @Override
            public List<Object> execute(RedisOperations operations) throws DataAccessException {
                Pool<Jedis> pool = null;
                Jedis jedis = null;
                Transaction tx = null;
                List<Object> result = null;
                for(int index = 1; index <= Const.RETRY_TIMES ; index++) {
                    try {
                        if (StringUtils.isNotBlank(lockCacheKey)) {
                            operations.watch(lockCacheKey);
                        }
                        if (enableTx) {
                            pool = RedisTemplateAdapter.getPool();
                            jedis = pool.getResource();
                            tx = jedis.multi();
                        }

                        redisBatchExecute.call(operations);
                        result = enableTx ? tx.exec() : operations.exec();
                    } catch (Exception e) {
                        if (index < Const.RETRY_TIMES) {
                            continue;
                        }
                        throw e;
                    } finally {
                        if(jedis.getClient().isInMulti()){
                            try {
                                tx.close();
                            } catch (IOException e) {
                                e.printStackTrace();
                            }
                        }
                        if (jedis.getClient().isInWatch()) {
                            operations.unwatch();
                        }
                        returnResource(pool, jedis);
//                        closeJedis(jedis);
                    }
                    break;
                }
                return result;
            }
        });
    }

    private static void returnResource(Pool<Jedis> pool, Jedis jedis){
        if (pool != null && jedis != null) {
            pool.returnResource(jedis);
        }
    }

    /**
     * 释放jedis资源(从Jedis 3.0开始使用jedis.close()方法释放资源,不再使用returnResource()函数)
     * @param jedis
     */
    private static void closeJedis(Jedis jedis){
//        jedis.quit();
//        try {
//            jedis.disconnect();
//        } catch (Exception e) {
//            throw e;
//        }
        if (jedis != null) {
            //starting from Jedis 3.0 returnResource method won't exist. Resouce cleanup should be done using @see {@link redis.clients.jedis.Jedis#close()
            jedis.close();
        }
    }
}